package day03;

import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.Calendar;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @Description: 字典状态变量
 * @Author ZYX
 * @Date 2022/2/10 19:00
 * @Version 1.0
 */
public class Demo08 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.addSource(new ClickSource())
                .keyBy(obj -> 1)
                .process(new KeyedProcessFunction<Integer, Event, String>() {
                    private MapState<String,Long> mapState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Long>("map键值对", Types.STRING,Types.LONG));
                    }

                    @Override
                    public void processElement(Event event, Context context, Collector<String> out) throws Exception {
                        if (mapState.contains(event.user)) {
                            mapState.put(event.user,mapState.get(event.user)+1L);
                        }else {
                            mapState.put(event.user,1L);
                        }

                        // 求PV平均值
                        long userNum = 0L;
                        long pvSum = 0L;

                        for (String user : mapState.keys()) {
                            userNum += 1L;
                            pvSum += mapState.get(user);
                        }
                        out.collect("当前pv的平均值是："+ (double)pvSum/userNum);
                    }
                }).print();

        env.execute();
    }

    public static class ClickSource implements SourceFunction<Event> {
        private boolean running = true;
        private String[] userArr = {"Alen","Bob","Carter","Dave","Eden","Frank","Gucci"};
        private String[] urlArr = {"./home","./cart","./fav","./user/1","./product/2"};
        private Random random = new Random();

        @Override
        public void run(SourceContext<Event> sourceContext) throws Exception {
            while (running){
                // collect 方法，向下游发送数据
                sourceContext.collect(
                        new Event(userArr[random.nextInt(userArr.length)],
                                urlArr[random.nextInt(urlArr.length)],
                                Calendar.getInstance().getTimeInMillis())
                );
                TimeUnit.SECONDS.sleep(1L);
            }
        }

        @Override
        public void cancel() {
            running = false;

        }
    }

    public static class Event{
        public String user;
        public String url;
        public Long timeStamp;

        public Event() {
        }

        public Event(String user, String url, Long timeStamp) {
            this.user = user;
            this.url = url;
            this.timeStamp = timeStamp;
        }

        @Override
        public String toString() {
            return "Event{" +
                    "user='" + user + '\'' +
                    ", url='" + url + '\'' +
                    ", timeStamp=" + timeStamp +
                    '}';
        }
    }

}
